package main

import (
	"bufio"
	"flag"
	"fmt"
	"io"
	"log"
	"math/rand"
	"net"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"
)

var (
	mode = flag.String("m", "", "server or client")
)

// tgpl section 8.2
/////////////////////// 同步server
func serverClock1() {
	listner, err := net.Listen("tcp", "localhost:8000")
	if err != nil {
		log.Fatal(err)
	}
	for {
		conn, err := listner.Accept()
		if err != nil {
			log.Print(err)
			continue // connection aborted
		}
		// 由于是同步，无法接收多个client连接
		handleConn(conn)
	}
}
func handleConn(c net.Conn) {
	defer c.Close()
	for {
		_, err := io.WriteString(c, time.Now().Format("15:01:05\n"))
		if err != nil {
			return // client 可能断开了,这时候server写出错
		}
		time.Sleep(1 * time.Second)
	}
}

///////////// 另外一种handleConn, read line by line
func handleConn_line_by_line(c net.Conn) {
	fmt.Printf("servring %s\n", c.RemoteAddr().String())
	for {
		data, err := bufio.NewReader(c).ReadString('\n')
		if err != nil {
			fmt.Println(err)
			return
		}
		tmp := strings.TrimSpace(string(data))
		if tmp == "STOP" {
			break
		}
		res := strconv.Itoa(rand.Int()) + "\n"
		c.Write([]byte(res))
	}
}

//////// 用nc, telent也可以模拟
func clientNetcat1() {
	conn, err := net.Dial("tcp", "localhost:8000")
	if err != nil {
		log.Fatal(err)
	}

	defer conn.Close()

	mustCopy := func (dst io.Writer, src io.Reader)  {
		// 直到src碰到EOF，或者错误发生才返回
		if _, err := io.Copy(dst, src); err != nil {
			log.Fatal(err)
		}
	}
	mustCopy(os.Stdout, conn)
}

/// 8.4.2 在channel关闭后，所有的send会panic，recv对closed channel会一直收到 0 
func badPipeline1() {
	// unbuffered channel 的特征是，当没有谁在recv时，send也会被block
	naturals := make(chan int)
	squares := make(chan int)
	go func ()  {
		for x := 1; x <= 4; x++ {
			naturals <- x
		}
		fmt.Println("naturals done")
		// 如果这里调了close,squares drain之后会一直收到 0 ！！
		close(naturals)
	}()

	go func ()  {
		for {
			x := <-naturals
			squares <- x*x
		}
		// 永远不会退出
		fmt.Println("squares done")
	}()

	for {
		fmt.Println(<-squares)
		time.Sleep(time.Second)
	}
}

func goodPipeline2() {
	naturals := make(chan int)
	squares := make(chan int)
	go func ()  {
		for x := 1; x <= 4; x++ {
			naturals <- x
		}
		fmt.Println("naturals done")
		// 当需要显式告知channel要关闭时，就要关闭，如果不是明确告知下游可以不手动关闭，GC会负责回收
		// 文件不适用这种情况，文件用完了必须 关闭
		close(naturals)
	}()

	go func ()  {
		for x := range naturals {
			squares <- x*x
		}
		// range退出时，表明上游channel已经关闭（收到上游通知了）
		close(squares)
	}()

	
	for x := range squares {
		fmt.Println(x)
	}
	fmt.Println("main ends")
}

/////////// 单向channel ,用以只能收或只能发的语义
func unidirectionalPipeline3() {
	counter := func (out chan<- int)  {
		for x:=0; x<4; x++ {
			out <- x
		}
		// 写完了就关闭
		close(out)
	}

	squarer := func (out chan<- int, in <-chan int)  {
		for v := range in {
			out <- v
		}
		close(out)
	}

	printer := func (in <-chan int)  {
		for v := range in {
			fmt.Println(v)
		}
	}

	naturals := make(chan int)
	square := make(chan int)
	///// 允许单向channel传参给双向，不能反过来!!
	go counter(naturals)
	go squarer(square, naturals)
	printer(square)
}

// 8.4.4
// buffer的好处在于使send, recv解耦
// leaked goroutine 不会被回收，所以要确保不需要了就停止
func bufferedChannel() {
	ch := make(chan int)
	// len是不准的，在并发代码里变更很快
	fmt.Printf("cap: %v, len: %v", cap(ch), len(ch))
}

//////////// 8.5 looping in parallel
func imageFile(f string) (string, error) { return "/tmp/a.txt", nil }

// 称完成独立的子问题叫 embarrassing parallel，这种问题是最容易改造成并行的
func makeThumbnails4(filenames []string) error {
	errors := make(chan error)
	for _, f := range filenames {
		go func(f string) {
			_, err := imageFile(f)
			errors <- err
		}(f) // 要在这里给f参数，不能直接给进 closure里 ！ 很可能当goroutine执行时，f 已经更新了多次
	}

	for range filenames {
		if err := <- errors; err != nil {
			// 当第一个non-nil err返回 时，就return，导致没有drain the errors channel，导致剩余worker都卡住了 goroutine leak
			return err 
		}
	}
	return nil
}

func makeThumbnails5(filenames []string) (thumbfiles []string, err error) {
	type item struct {
		thumbfile string
		err error
	}

	// 使用buffered channel来避免 drain 问题
	ch := make(chan item, len(filenames))
	for _, f := range filenames {
		go func(f string) {
			var it item
			it.thumbfile, it.err = imageFile(f)
			ch <- it  // 不会block worker goroutine
		}(f)
	}

	for range filenames {
		it := <- ch
		if it.err != nil {
			// 其中一个worker返回 err后，其它worker还在跑，设计有点不太好
			return nil, it.err
		}
		thumbfiles = append(thumbfiles, it.thumbfile)
	}
	return thumbfiles, nil
}

// 返回创建的文件的总大小
func makeThumbnails6(filenames <-chan string) int64 {
	sizes := make(chan int64)
	var wg sync.WaitGroup
	for f := range filenames {
		log.Printf("file: %v", f)
		wg.Add(1) // 必须在开启goroutine前调用
		go func(f string) {
			defer wg.Done() // 相当于 -1
			thumb, err := imageFile(f)
			if err != nil {
				log.Println(err)
				return // 和makeThumbnails5 不一样，没有把err 传给outer goroutine
			}
			info, _ := os.Stat(thumb)
			//log.Printf("%v worker begins write result to channel", f)
			//当没有谁在recv时，send也会被block
			sizes <- info.Size()
			//log.Printf("%v worker ends", f)
		}(f)
	}

	// closer
	go func() {
		// 若把Wait 放在主goroutine loop前面，则Wait一直会卡在那；因为，没有谁在recv sizes channel
		// 若放在main goroutine loop后面，永远不会执行它，因为sizes channel不会被关闭
		wg.Wait()
		log.Println("ready to close sizes channel")
		close(sizes)
	}()

	var total int64
	for s := range sizes {
		log.Printf("size: %v", s)
		total += s
	}
	return total
}

func tLoopingInParallel() {
	ch := make(chan string, 2)
	ch <- "a"
	ch <- "b"
	close(ch)
	makeThumbnails6(ch)
}
///////////////////////

func main() {
	flag.Parse()
	if *mode == "s" {
		serverClock1()
	} else if *mode == "c" {
		clientNetcat1()
	}
	//badPipeline1()
	//goodPipeline2()
	//unidirectionalPipeline3()
	//tLoopingInParallel()
}